package com.nc.mem.serial.graph.container;

import java.lang.reflect.Modifier;
import java.text.MessageFormat;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.nc.mem.serial.Serializer;
import com.nc.mem.serial.graph.boot.ProvisionService;
import com.nc.mem.serial.graph.boot.ProvisionServiceFactory;
import com.nc.mem.serial.graph.boot.RegistrationListener;
import com.nc.mem.serial.graph.common.CompatibleSerializer;

public final class SerializerRegistry {

	static final HashMap<String, SerializerRegistry> ALL_REPOS = new HashMap<String, SerializerRegistry>(2);

	public static SerializerRegistry getInstance() {
		return getInstance(null, Boolean.valueOf(System.getProperty("c.n.m.s.g.c.SR.compatibleFallback", "true")));
	}

	public static SerializerRegistry getInstance(final boolean compatibleFallback) {
		return getInstance(null, compatibleFallback);
	}

	/**
	 * @param context
	 * @param compatibleFallback
	 * @return
	 */
	public static SerializerRegistry getInstance(final String context, final boolean compatibleFallback) {
		final String key = context == null || context.trim().isEmpty() ? null : context;

		SerializerRegistry repo = ALL_REPOS.get(key);
		if (repo == null) {
			ALL_REPOS.put(key, repo = new SerializerRegistry(ProvisionServiceFactory.resolveOrFallback(key), compatibleFallback));
			repo.finishConstruction();
		} else {
			if (repo.compatibleFallback != compatibleFallback) {
				repo.getLogger().warn("Attempted to recreate repository with distinct imcompatibleFallback!!!");
			}
		}
		return repo;
	}

	private final ConcurrentHashMap<Class<?>, Integer> classToId;
	private final ConcurrentHashMap<Integer, Class<?>> idToClass;
	private final ConcurrentHashMap<Class<?>, Serializer> serializersForCompat;
	private final ConcurrentHashMap<Class<?>, Serializer> serializers;
	private final IdentityHashMap<Class<?>, Serializer> altSerializers;
	private final IdentityHashMap<Class<?>, Void> abstractionSet;
	private final IdentityHashMap<Class<?>, Void> fullgraphSet;
	private final IdentityHashMap<Class<?>, Void> closures;

	private final Iterable<RegistrationListener> listeners;
	private final AtomicInteger seq;
	private final boolean frozen;
	private final boolean compatibleFallback;
	transient volatile Iterable<ProvisionService> services;

	public SerializerRegistry(final Iterable<ProvisionService> services, final boolean compatibleFallback) {
		this.services = services;
		classToId = new ConcurrentHashMap<Class<?>, Integer>();
		idToClass = new ConcurrentHashMap<Integer, Class<?>>();
		fullgraphSet = new IdentityHashMap<Class<?>, Void>();
		serializersForCompat = new ConcurrentHashMap<Class<?>, Serializer>();
		serializers = new ConcurrentHashMap<Class<?>, Serializer>(64);
		altSerializers = new IdentityHashMap<Class<?>, Serializer>(2);
		abstractionSet = new IdentityHashMap<Class<?>, Void>(64);
		closures = new IdentityHashMap<Class<?>, Void>(2);
		this.compatibleFallback = compatibleFallback;

		boolean f = false;
		int totalMappings = 0;
		Set<RegistrationListener> allListeners = null;

		// Step 1 - register mappings, replacements and listeners
		if (services != null) {
			// verify and register listeners

			for (final ProvisionService ps : services) {

				getLogger().info("Registering definitions for {}", ps.getClass().getSimpleName());

				final Iterable<RegistrationListener> listeners = ps.listeners();
				if (listeners != null) {
					allListeners = new HashSet<RegistrationListener>(2);

					for (final RegistrationListener listener : listeners) {
						if (listener != null) {
							allListeners.add(listener);
						}
					}
					if (allListeners.size() == 0) {
						allListeners = null;
					}
				}

				final Iterable<Entry<Integer, Class<?>>> mappings = ps.installableMappings();

				if (mappings != null) {
					for (final Entry<Integer, Class<?>> e : mappings) {
						registerClass(e.getKey(), e.getValue(), false);
					}
				}
			}

			totalMappings = getTotalMappings();
			f = totalMappings > 0;
		}

		listeners = allListeners;
		frozen = f;

		if (f) {
			seq = null;
			getLogger().info("{} booted in frozen mode! Total mappings: {}", getClass().getSimpleName(), totalMappings);
		} else {
			biRegisterUncond(Object.class, 0);
			biRegisterUncond(Enum.class, 1);
			seq = new AtomicInteger(2);
			getLogger().info("{} booted in non-frozen mode!", getClass().getSimpleName());
		}

	}

	void biRegisterUncond(final Class<?> clazz, final int id) {
		classToId.put(clazz, id);
		idToClass.put(id, clazz);
	}

	void broadCastAll() {
		if (listeners != null) {
			final Iterable<Entry<Integer, Class<?>>> es = idToClassMappings();

			for (final Entry<Integer, Class<?>> e : es) {
				for (final RegistrationListener l : listeners) {
					l.onClassRegistered(e.getValue(), e.getKey());
				}
			}
		}
	}

	private void broadCastClassRegistered(final Class<?> clazz, final Integer id) {
		final Iterable<RegistrationListener> listeners = this.listeners;
		if (listeners != null) {
			for (final RegistrationListener listener : listeners) {
				listener.onClassRegistered(clazz, id);
			}
		}
	}

	public Integer classToId(final Class<?> clazz) {
		return classToId.get(clazz);
	}

	private Integer decideNewIdAndBroadCast(final Class<?> clazz) {
		Integer ret = null;

		if (clazz.isEnum()) {
			ret = classToId(Enum.class);
			registerIfAbsent(clazz, ret);
		} else if (clazz.isInterface() || Modifier.isAbstract(clazz.getModifiers())) {
			ret = -1;
		} else if (isClosure(clazz)) {
			ret = registerClosureImpl(clazz);
		} else {
			if (frozen) {
				if (compatibleFallback) {
					ret = registerForCompatibility(clazz);
				} else {
					throw new UnsupportedOperationException("Representation is FROZEN and compatible fallback is turned off!!! You must statically map " + clazz);
				}
			} else {
				ret = seq.incrementAndGet();
				final Integer old = registerIfAbsent(clazz, ret);
				if (old != null) {
					ret = old;
				}
				registerUncond(ret, clazz);
			}
		}
		broadCastClassRegistered(clazz, ret);
		return ret;
	}

	public void deregisterForFullGraph(final Class<?>[] classes) {
		for (final Class<?> c : classes) {
			fullgraphSet.remove(c);
		}

	}

	void finishConstruction() {

		try {
			// Step 2 - BroadCast everything that has been mapped. We call this separately to allow
			// the constructor to return, otherwise we'd be notifying listeners with a
			// half-constructed object.
			broadCastAll();

			// Step 3 - After mappings have been registered, we can install for serializers for
			// them.
			// This will enforce frozen mode semantics.
			final Iterable<ProvisionService> services = this.services;
			if (services != null) {
				for (final ProvisionService ps : services) {
					final Iterable<Entry<Class<?>, Serializer>> is = ps.installableSerializers();

					if (is != null) {
						for (final Entry<Class<?>, Serializer> e : is) {
							register(e.getKey(), e.getValue());
						}
					}

					final Iterable<Entry<Serializer, Iterable<Class<?>>>> abstractions = ps.installableAbstractions();

					if (abstractions != null) {
						for (final Entry<Serializer, Iterable<Class<?>>> e : abstractions) {
							final Serializer key = e.getKey();
							final Iterable<Class<?>> value = e.getValue();
							if (value != null) {
								for (final Class<?> c : value) {
									if (c != null) {
										registerForParentAbstraction(c, key);
									}
								}
							}
						}
					}

					final Iterable<Class<?>> closures = ps.installableClosures();

					if (closures != null) {
						for (final Class<?> clazz : closures) {
							registerClosure(clazz);
						}
					}
				}
			}
		} catch (final Throwable e) {
			throw new ExceptionInInitializerError(e);
		} finally {
			this.services = null;
		}
	}

	public Serializer getAlternative(final Class<?> clazz) {
		return altSerializers.get(clazz);
	}

	public Serializer getForCompatiblity(final Class<?> clazz) {
		return serializersForCompat.get(clazz);
	}

	Logger getLogger() {
		return LoggerFactory.getLogger(getClass());
	}

	public int getTotalMappings() {
		return idToClass.size();
	}

	public boolean hasAlternative(final Class<?> c) {
		return altSerializers.containsKey(c);
	}

	public Class<?> idToClass(final int id) {

		return idToClass.get(id);
	}

	Iterable<Entry<Integer, Class<?>>> idToClassMappings() {
		return idToClass.entrySet();
	}

	public boolean isAbstracted(final Class<?> clazz) {
		return abstractionSet.containsKey(clazz);
	}

	public boolean isClosure(final Class<?> clazz) {
		for (final Class<?> key : closures.keySet()) {
			if (key.isAssignableFrom(clazz)) {
				return true;
			}
		}
		return false;
	}

	public boolean isForCompatibility(final Class<?> clazz) {
		return classToId.get(Object.class).equals(classToId.get(clazz));
	}

	public boolean isFullGraph(final Class<?> clazz) {
		return fullgraphSet.containsKey(clazz);
	}

	public Map<Integer, List<Class<?>>> orderedView() {
		final Set<Entry<Class<?>, Integer>> set = classToId.entrySet();

		final Map<Integer, List<Class<?>>> m = new TreeMap<Integer, List<Class<?>>>();
		for (final Entry<Class<?>, Integer> entry : set) {
			List<Class<?>> list = m.get(entry.getValue());

			if (list == null) {
				list = new LinkedList<Class<?>>();
				m.put(entry.getValue(), list);
			}
			list.add(entry.getKey());
		}

		return m;

	}

	public void putForCompatiblity(final Class<?> clazz, final Serializer ser) {
		serializersForCompat.put(clazz, ser);
	}

	public void register(final Class<?> c, final Serializer s) {
		if (s == null) {
			throw new IllegalArgumentException();
		}

		final int id = registerClass(c);

		if (c != Object.class && id == classToId(Object.class)) {
			serializersForCompat.put(c, s);
			serializers.put(c, CompatibleSerializer.INSTANCE);
		} else {
			serializers.put(c, s);
		}

	}

	public void register(final Class<?>[] classes, final Serializer s) {
		for (final Class<?> clazz : classes) {
			register(clazz, s);
		}
	}

	public boolean registerAlternative(final Class<?> c, final Serializer s) {
		if (s == null) {
			throw new IllegalArgumentException();
		}
		return altSerializers.put(c, s) == null;
	}

	public void registerAlternatives(final Class<?>[] classes, final Serializer s) {
		for (final Class<?> c : classes) {
			registerAlternative(c, s);
		}
	}

	public int registerClass(final Class<?> clazz) {
		Integer ret = classToId.get(clazz);
		if (ret == null) {
			ret = decideNewIdAndBroadCast(clazz);
		}
		return ret;
	}

	public void registerClass(final Integer id, final Class<?> clazz) {
		if (!clazz.isInterface() && !Modifier.isAbstract(clazz.getModifiers())) {
			if (classToId.putIfAbsent(clazz, id) != null || idToClass.putIfAbsent(id, clazz) != null) {
				throw new IllegalArgumentException(MessageFormat.format("{0} already registered with id {1}", clazz, id));
			}
		}
	}

	private void registerClass(final Integer id, final Class<?> clazz, final boolean broadCast) {
		if (registerIfAbsent(clazz, id) != null || registerIfAbsent(id, clazz) != null) {
			throw new IllegalArgumentException(MessageFormat.format("{0} already registered with id {1}", clazz, id));
		}
		if (broadCast) {
			broadCastClassRegistered(clazz, id);
		}
	}

	public final void registerClosure(final Class<?> closure) {
		final Integer id = classToId.get(closure);

		if (id != null) {
			throw new IllegalAccessError(MessageFormat.format("Closure ids {0} must NOT be supplied!", closure.getName()));
		}

		closures.put(closure, null);
	}

	public int registerClosureImpl(final Class<?> closure) {
		for (final Class<?> clazz : closures.keySet()) {
			if (clazz.isAssignableFrom(closure)) {
				return registerForCompatibility(closure);
			}
		}

		throw new IllegalArgumentException();
	}

	public int registerForCompatibility(final Class<?> clazz) {
		final int pId = classToId.get(Object.class);

		final Integer old = registerIfAbsent(clazz, pId);
		if (old != null && !old.equals(pId)) {
			throw new IllegalArgumentException("Class has already been registered with id " + old);
		}

		getLogger().warn("Registering {} for compatibility!", clazz.getName());

		return pId;
	}

	public void registerForFullGraph(final Class<?>... classes) {
		final IdentityHashMap<Class<?>, Void> fgc = fullgraphSet;
		for (final Class<?> clazz : classes) {
			fgc.put(clazz, null);
		}
	}

	public void registerForParentAbstraction(final Class<?> c, final Serializer s) {
		abstractionSet.put(c, null);
		register(c, s);
	}

	Integer registerIfAbsent(final Class<?> clazz, final int id) {
		return classToId.putIfAbsent(clazz, id);
	}

	Class<?> registerIfAbsent(final int id, final Class<?> clazz) {
		return idToClass.putIfAbsent(id, clazz);
	}

	public int registerLikeParent(final Class<?> clazz, final Class<?> parent) {
		int pId = registerClass(parent);

		if (pId == -1) {
			if (frozen) {
				throw new IllegalArgumentException("Parent " + parent + " must be registered!");
			}
			pId = seq.incrementAndGet();
			final Integer old = registerIfAbsent(parent, pId);
			if (old != null) {
				pId = old;
			}
			registerUncond(pId, parent);
		}

		registerIfAbsent(clazz, pId);

		return pId;
	}

	void registerUncond(final Class<?> clazz, final int id) {
		classToId.put(clazz, id);
	}

	void registerUncond(final int id, final Class<?> clazz) {
		idToClass.put(id, clazz);
	}

	public void removeCompatible(final Class<?> clazz) {
		serializersForCompat.remove(clazz);
	}

	public Serializer serializer(final Class<?> clazz) {
		return serializers.get(clazz);
	}
}
